package com.innodata.riskEngine.mock;

import com.innodata.riskEngine.common.pojo.ChannelDataPO;
import com.innodata.riskEngine.common.pojo.SourcePO;
import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * @Author liboshuai
 * @Date 2023/11/2 23:19
 * Flink风控预警简易批处理版本
 */
@Slf4j
public class FlinkRiskRule {
    private static final String BANK_NAME = "Test Bank";

    /**
     * 游戏抽奖预警
     *
     * @param sourcePOs 事件列表
     * @param threshold 阈值
     * @param timeFrame 时间范围
     */
    public static List<String> gameLotteryWarning(List<SourcePO> sourcePOs, long threshold, long timeFrame) {
        Map<String, Long> lotteryCounts = new HashMap<>();
        long currentTime = System.currentTimeMillis();
        List<String> messageList = new ArrayList<>();

        for (SourcePO po : sourcePOs) {
            if (checkSourcePO(po)) {
                continue;
            }
            if (checkChannelPO(po.getChannelData())) {
                continue;
            }
            // Skip the records out of the window size
            if (currentTime - po.getEventTimestamp() > timeFrame) {
                continue;
            }

            String userId = po.getChannelData().getUserId();
            Long lotteryNumber = po.getChannelData().getLotteryNumber();

            lotteryCounts.put(userId, lotteryCounts.getOrDefault(userId, 0L) + lotteryNumber);
            if (lotteryCounts.get(userId) > threshold) {
                messageList.add(String.format("[异常高频抽奖]%s: 活动名称%s中游戏用户%s最近%d分钟内抽奖数据为%d，超过%d次，请您及时查看原因！",
                        BANK_NAME, po.getChannelData().getCampaignId(), userId,
                        TimeUnit.MILLISECONDS.toMinutes(timeFrame), lotteryCounts.get(userId), threshold));
            }
        }
        return messageList;
    }

    private static boolean checkChannelPO(ChannelDataPO channelData) {
        if (Objects.isNull(channelData)) {
            log.warn("channelData is null");
            return true;
        }
        String campaignId = channelData.getCampaignId();
        if (Objects.isNull(campaignId)) {
            log.warn("campaignId is null");
            return true;
        }
        String userId = channelData.getUserId();
        if (Objects.isNull(userId)) {
            log.warn("userId is null");
            return true;
        }
        Long lotteryNumber = channelData.getLotteryNumber();
        if (Objects.isNull(lotteryNumber) || lotteryNumber < 1) {
            log.warn("lotteryNumber is null or less than 1");
            return true;
        }
        return false;
    }

    private static boolean checkSourcePO(SourcePO sourcePO) {
        String channel = sourcePO.getChannel();
        if (!Objects.equals(channel, "game")) {
            log.warn("channel is not game");
            return true;
        }
        String bank = sourcePO.getBank();
        if (Objects.isNull(bank)) {
            log.warn("bank is null");
            return true;
        }
        long eventTimestamp = sourcePO.getEventTimestamp();
        if (eventTimestamp <= 0) {
            log.warn("eventTimestamp is less than 0");
            return true;
        }
        return false;
    }
}
